package com.atguigu.chapter07.c_timer;

import com.atguigu.chapter5.source.WaterSensor;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

/**
 * @ClassName: Flink14_Timeer_Event
 * @Description:
 * @Author: kele
 * @Date: 2021/4/7 13:16
 *
 * 监控水位传感器的水位值，如果水位值在5s(processing time)连续上升，则报警
 *
 **/
public class Flink16_Timeer_Project {

    public static void main(String[] args) {

        Configuration conf = new Configuration();
        conf.setInteger("rest.port",20000);

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(1);

        env.socketTextStream("hadoop162",8888)
                .map(datas -> {

                    String[] data = datas.split(",");
                    return new WaterSensor(data[0],
                            Long.parseLong(data[1]) * 1000,
                            Integer.valueOf(data[2]));
                })
                //添加水印
               .assignTimestampsAndWatermarks(WatermarkStrategy
                       .<WaterSensor>forMonotonousTimestamps()
                       .withTimestampAssigner(new SerializableTimestampAssigner<WaterSensor>() {
                           @Override
                           public long extractTimestamp(WaterSensor element, long recordTimestamp) {
                               return element.getTs();
                           }
                       })
               )
                .keyBy(WaterSensor::getId)
                /**
                 * 使用触发器，
                 *  1、开始注册一个触发器
                 *  2、如果这次相较于上次下降，则取消触发器，重新创建一个触发器
                 *  3、如果上升，且触发器的时间超过5s，则报警
                 */
                .process(new KeyedProcessFunction<String, WaterSensor,String>() {

                    boolean isfirst = true;
                    Integer lastVC = 0;
                    Long ts ;

                    @Override
                    public void processElement(WaterSensor value, Context ctx, Collector<String> out) throws Exception {

                        if(isfirst){
                            ts = ctx.timestamp()+5000;

                            ctx.timerService().registerEventTimeTimer(ts);

                            isfirst = false;

                        }else {

                            if(lastVC >= value.getVc()){
                                ctx.timerService().deleteEventTimeTimer(ts);

                                ts = ctx.timestamp()+5000;
                                ctx.timerService().registerEventTimeTimer(ts);
                            }
                        }
                        lastVC = value.getVc();

                    }

                    //触发时间到了,
                    @Override
                    public void onTimer(long timestamp, OnTimerContext ctx, Collector<String> out) throws Exception {

                        out.collect(ctx.getCurrentKey()+"   5s内连续上升，报警......");

                        //触发了该报警，需要重新注册，可以在下次重新注册
                        isfirst = true;

                    }
                })
                .print();



        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }


    }

}
